package main

import (
	"log"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	// cluster "github.com/bsm/sarama-cluster"
)

type Producer struct{
	producer sarama.AsyncProducer
}

func NewProducer() (Producer, error) {
	// log
	// sarama.Logger = logger

	// config
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 60 * time.Second
	// config.Producer.RequiredAcks = sarama.WaitForAll
	// config.Producer.Partitioner = sarama.NewRandomPartitioner

	p, err := sarama.NewAsyncProducer(strings.Split("192.168.134.132:9092", ","), config)
	// producer, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)
	return Producer{producer:p}, err
}

func (p Producer) ProduceMsg(args Args) error {

	// defer producer.Close()

	// produce msg
	msg := &sarama.ProducerMessage{
		Timestamp: time.Now(),
		Topic: "test",
		Key:   sarama.StringEncoder(args.ip),
		Value: sarama.ByteEncoder(args.data),
	}

	p.producer.Input() <- msg

	select {
	case err := <-p.producer.Errors():
		log.Printf("[Error] Send message error: %v\n", err)
	case msg := <-p.producer.Successes():
		key, _ := msg.Key.Encode()
		val, _ := msg.Value.Encode()
		timestamp := msg.Timestamp//.Format("2006-01-02 15:04:05")
		logmsg := "[Info] Send message success. Topic:%v, Key:%v, Value:%v, Timestamp:%v\n"
		log.Printf(logmsg, msg.Topic, string(key), string(val), timestamp)
	}

	return nil
}

func (p Producer) Close() error {
	return p.producer.Close()
}
